package com.jc.demo.zk;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Random;

import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE;

public class Master implements Watcher {
    ZooKeeper zk;
    String hostPort;

    String serverId = Long.toString(new Random().nextLong());
    public boolean isLeader = false;

    Master(String hostPort) {
        this.hostPort = hostPort;
    }

    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort, 15000, this);
    }

    public void process(WatchedEvent e) {
        System.out.println(e);
    }

    public void stopZK() throws InterruptedException {
        zk.close();
    }


    void runFormaster() throws InterruptedException {
        while (true) {
            try {
                zk.create("/master", serverId.getBytes(), OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                isLeader = true;
                break;
            } catch (KeeperException.NodeExistsException e) {
                isLeader = false;
                break;
            } catch (KeeperException.ConnectionLossException e) {

            } catch (KeeperException e) {
            }

            if (checkMaster()) break;

        }
    }

    private boolean checkMaster() throws InterruptedException {
        while (true) {
            Stat stat = new Stat();
            byte data[] = new byte[0];
            try {
                data = zk.getData("/master", false, stat);
            } catch (KeeperException.NodeExistsException e) {
                return false;
            } catch (KeeperException.ConnectionLossException e) {

            } catch (KeeperException e) {
                e.printStackTrace();
            }
            isLeader = new String(data).equals(serverId);
            return true;
        }
    }

    public static void main(String args[])
            throws Exception {
        Master m = new Master("vm1.jevoncode.com:2181,vm2.jevoncode.com:2181,vm3.jevoncode.com:2181");
        m.startZK();

        m.runFormaster();

        if (m.isLeader) {
            System.out.println("I'm the leader");
            Thread.sleep(60000);
        } else {
            System.out.println("Someone else is the leader");
        }

        m.stopZK();
    }
}